package com.sparrow.common.schedule.ownerscheduler;

import com.sparrow.common.schedule.queue.FixTimeDelayQueue;
import org.jetbrains.annotations.NotNull;

import java.util.Queue;
import java.util.concurrent.*;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

/**
 * worker线程负责接受addQueue的任务 转移到taskQueue
 * 调度执行延迟转移 -> worker 执行任务
 */
public class OwnerScheduler2 {
    private ScheduledExecutorService executor;

    private final Queue<MiniScheduledTask> taskQueue = new FixTimeDelayQueue();
    private volatile ScheduledFuture<?> currentFuture;
    private volatile boolean isShutdown = false;
    private SchedulerOwner owner;

    private Thread worker;

    private ConcurrentLinkedQueue<MiniScheduledTask> addQueue = new ConcurrentLinkedQueue<>();
    private ConcurrentLinkedQueue<MiniScheduledTask> runQueue = new ConcurrentLinkedQueue<>();

    private void loop() {
        int loop = 0;
        while (!isShutdown) {
            int size = addQueue.size();
            for (int i = 0; i < size; i++) {
                MiniScheduledTask poll = addQueue.poll();
                if (poll == null) {
                    break;
                }
                addTask(poll);
                loop = 0;
            }

            size = runQueue.size();
            for (int i = 0; i < size; i++) {
                MiniScheduledTask poll = runQueue.poll();
                if (poll == null) {
                    break;
                }
                runTask(poll);
                loop = 0;
            }

            scheduleNext();

            loop ++;
            if (loop == 100) {
                loop = 0;
                try {
                    MILLISECONDS.sleep(50);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    public OwnerScheduler2(ScheduledExecutorService executor, SchedulerOwner owner) {
        this.executor = executor;
        this.owner = owner;
        worker = new Thread(() -> {
            loop();
        });
        worker.setDaemon(false);
        worker.start();
    }

    public void schedule(Runnable task, long delayMillis) {
        schedule(task, delayMillis, 0);
    }

    public void scheduleAtFixedRate(Runnable task, long initialDelay, long intervalMillis) {
        if (intervalMillis <= 0) {
            throw new IllegalArgumentException("intervalMillis must be > 0");
        }
        schedule(task, initialDelay, intervalMillis);
    }

    private void schedule(Runnable task, long delayMillis, long intervalMillis) {
        if (isShutdown) {
            return;
        }
        MiniScheduledTask scheduledTask = new MiniScheduledTask(task, System.currentTimeMillis() + delayMillis, intervalMillis);
        addQueue.offer(scheduledTask);
    }

    private void addTask(MiniScheduledTask scheduledTask) {
        taskQueue.offer(scheduledTask);

        if (currentFuture != null) {
            currentFuture.cancel(false);
            currentFuture = null;
        }

        scheduleNext();
    }

    public void setShutdown(boolean shutdown) {
        isShutdown = shutdown;
    }

    public synchronized void shutdown() {
        isShutdown = true;
        if (currentFuture != null) {
            currentFuture.cancel(false);
            currentFuture = null;
        }
        taskQueue.clear();
    }

    private void scheduleNext() {
        MiniScheduledTask next = taskQueue.peek();
        if (next == null) {
            return;
        }

        long now = System.currentTimeMillis();
        long delay = Math.max(0, next.executeTime - now);
        currentFuture = executor.schedule(() -> {
            MiniScheduledTask polled = taskQueue.poll();
            if (polled != next) {
                // 任务错乱，不执行
                //重新加回队列
                addTask(polled);
                return;
            }
            runQueue.offer(next);
        }, delay, TimeUnit.MILLISECONDS);
    }

    private void runTask(MiniScheduledTask next) {
        try {
            next.runnable.run();
        } catch (Exception e) {
            System.err.println("Error in scheduled task for entity " + owner);
            e.printStackTrace();
        }
        onTaskFinished(next);
    }

    private void onTaskFinished(MiniScheduledTask task) {
        currentFuture = null;
        if (!isShutdown) {
            if (task.isLoop()) {//循环执行的任务需要从新放到任务队列
                long nextTime = System.currentTimeMillis() + task.intervalMillis;
                task.setExecuteTime(nextTime);
                addTask(task);
            }
            scheduleNext();
        }
    }

    private static class MiniScheduledTask implements Delayed {
        final Runnable runnable;
        long executeTime; // 绝对执行时间戳 ms
        final long intervalMillis; // 0 表示一次性， >0 表示周期

        MiniScheduledTask(Runnable runnable, long executeTime, long intervalMillis) {
            this.runnable = runnable;
            this.executeTime = executeTime;
            this.intervalMillis = intervalMillis;
        }

        public boolean isLoop() {
            return intervalMillis > 0;
        }

        public long getExecuteTime() {
            return executeTime;
        }

        @Override
        public long getDelay(@NotNull TimeUnit unit) {
            return executeTime - System.currentTimeMillis();
        }

        public int compareTo(Delayed other) {
            if (other == this) // compare zero if same object
                return 0;
            if (other instanceof MiniScheduledTask) {
                MiniScheduledTask x = (MiniScheduledTask)other;
                long diff = executeTime - x.executeTime;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
                else
                    return 0;
            }
            long diff = getDelay(MILLISECONDS) - other.getDelay(MILLISECONDS);
            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
        }

        public void setExecuteTime(long executeTime) {
            this.executeTime = executeTime;
        }
    }

}
